package com.nanohadoop.mapreduce.core;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Mapper 接口定义了 MapReduce 框架中 map 阶段的基本操作
 * 用户通过继承该类并实现 map 方法来定义具体的 map 逻辑
 * 添加了输入记录计数
支持进度报告和计数器
提供了 WordCount 示例实现
 */
public abstract class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

    private MapperContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context;
    private final AtomicLong mapInputRecords = new AtomicLong(0);
    private final AtomicLong mapOutputRecords = new AtomicLong(0);

    /**
     * 设置运行上下文
     */
    public void setContext(MapperContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) {
        this.context = context;
    }

    /**
     * 初始化方法,在map任务开始前调用
     */
    protected void setup(MapperContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) throws IOException, InterruptedException {
        // 默认空实现,用户可以重写此方法进行初始化操作
    }

    /**
     * map方法,处理一条输入记录
     * @param key 输入key
     * @param value 输入value
     * @param context 用于输出结果的上下文
     */
    public abstract void map(KEYIN key, VALUEIN value, MapperContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context)
            throws IOException, InterruptedException;

    /**
     * 清理方法,在map任务结束后调用
     */
    protected void cleanup(MapperContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) throws IOException, InterruptedException {
        // 默认空实现,用户可以重写此方法进行清理操作
    }

    /**
     * run方法封装了map任务的执行流程
     */
    public void run(MapperContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> context) throws IOException, InterruptedException {
        setup(context);
        try {
            while (context.nextKeyValue()) {
                mapInputRecords.incrementAndGet();
                map(context.getCurrentKey(), context.getCurrentValue(), context);
            }
        } finally {
            cleanup(context);
        }
    }

    /**
     * 获取处理的输入记录数
     */
    public long getMapInputRecords() {
        return mapInputRecords.get();
    }

    /**
     * 获取输出的记录数
     */
    public long getMapOutputRecords() {
        return mapOutputRecords.get();
    }


}